package com.heima.stream;

import cn.hutool.core.convert.Convert;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
@Slf4j
public class WordCountsDemo3 {

    /* 每隔15秒钟输出一次过去25秒内topicA里的word count，结果写入到TopicB */
    @Bean
    public KTable<Windowed<Object>, Long> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        KTable<Windowed<Object>, Long> countKtable = stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) {
                return Arrays.asList(value.toString().split(" "));
            }
        }).map(new KeyValueMapper<String, Object, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(String key, Object value) {
                System.out.println("key:::: "+key);
                System.out.println("value:::: "+value);
                return new KeyValue<String, String>(Convert.toStr(value), "1");
            }
        }).groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(25).toMillis()).advanceBy(Duration.ofSeconds(15).toMillis()))
                .count();
        countKtable.toStream().map(new KeyValueMapper<Windowed<Object>, Long, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(Windowed<Object> key, Long value) {
                System.out.println("key:::: "+key);
                System.out.println("value:::: "+value);
                return new KeyValue<String, String>(key.toString(), value.toString());
            }
        }).to("itcast-topic-out");

        return countKtable;
    }
}
